package run.liuli.tran.user.quartz;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.util.concurrent.ListenableFutureCallback;
import run.liuli.tran.user.enmu.ToCouponStatus;
import run.liuli.tran.user.pojo.UserCouponTran;
import run.liuli.tran.user.service.UserCouponTranService;

import javax.annotation.Resource;
import java.util.List;

public class CouponJob extends QuartzJobBean {

    @Resource
    private UserCouponTranService userCouponTranService;

    @Resource
    private KafkaTemplate<String,Object> kafkaTemplate;

    @Override
    protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {

        List<UserCouponTran> tranMessages = userCouponTranService.list(new QueryWrapper<UserCouponTran>()
                .eq("status", ToCouponStatus.USER_COUPON_NO_SEND.getCode()));

        tranMessages.forEach(tranMsg->{
            //通知发放代金券
            kafkaTemplate.send("coupon",tranMsg.getUid()+","+tranMsg.getCouponMoney()+","+tranMsg.getTid())
                    .addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                        @Override
                        public void onFailure(Throwable throwable) {
                            throw new RuntimeException("消息发送失败");
                        }

                        @Override
                        public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                            //更改事务记录状态
                            tranMsg.setStatus(ToCouponStatus.USER_COUPON_SEND.getCode());
                            userCouponTranService.updateById(tranMsg);
                        }
                    });
        });
    }
}
